package com.abel.bigwater.data.sanchuan

import com.abel.bigmeter.service.DataParam
import com.abel.bigwater.data.BcdString
import com.abel.bigwater.data.mapper.DataMapper
import com.abel.bigwater.data.mapper.RtuMapper
import com.abel.bigwater.model.BwData
import com.abel.bigwater.model.BwRtu
import com.abel.bigwater.model.BwRtuLog
import com.alibaba.fastjson.JSON
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.handler.timeout.IdleState
import io.netty.handler.timeout.IdleStateEvent
import io.netty.util.AttributeKey
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async
import org.joda.time.DateTime
import org.slf4j.LoggerFactory
import java.io.File
import java.util.*
import java.util.concurrent.locks.ReentrantReadWriteLock
import kotlin.collections.ArrayList
import kotlin.concurrent.withLock

open class SanchuanTcpHandler(val dataMapper: DataMapper, val rtuMapper: RtuMapper, val firmId: String, val firmCode: String)
    : ChannelInboundHandlerAdapter() {

    private val lock = ReentrantReadWriteLock()

    /**
     * 1个会话最多提取数据3次。
     */
    override fun channelActive(ctx: ChannelHandlerContext?) {
        // store fetchTimes
        val key = if (AttributeKey.exists(KEY_FETCH_TIMES))
            AttributeKey.valueOf<Int>(KEY_FETCH_TIMES)
        else
            AttributeKey.newInstance<Int>(KEY_FETCH_TIMES)
        val ft = ctx!!.channel().attr(key)

        ft.setIfAbsent(0)
    }

    /**
     * Calls [ChannelHandlerContext.fireChannelRead] to forward
     * to the next [ChannelInboundHandler] in the [ChannelPipeline].
     *
     * Sub-classes may override this method to change behavior.
     */
    override fun channelRead(ctx: ChannelHandlerContext?, msg: Any?) {
        lgr.info("GOT Mesage: $msg")

        if (msg is SanchuanPacket) {
            handlePacket(msg, ctx)
        }
    }

    /**
     * handle the san-chuan packet.
     * @param msg the san-chuan packet.
     * @param ctx the netty context
     */
    protected fun handlePacket(msg: SanchuanPacket, ctx: ChannelHandlerContext?) {
        val extId = "${firmCode}${msg.meterId!!.reversedBcdString()}"

        when (msg.ctrlb) {
            CtrlbType.Online -> {
                val data = msg.data as MeterOnlineData

                GlobalScope.async {
                    try {
                        storeOnline(msg, data, ctx)
                    } catch (ex: Exception) {
                        lgr.error("保存数据异常: ${ex.message}", ex)
                    }
                }

                // 0x00 表示服务器有后续命令， 水表不能下线
                replyPacket(msg, ctx)

                // 提取累计行度
                val range = dataMapper.realtimeDateRange(DataParam(extId = extId).apply {
                    sampleTime2 = Date()
                })
                fetchForward(msg, ctx, recentWeek(range?.maxDateTime))
            }

            CtrlbType.GprsOnline -> {
                GlobalScope.async {
                    try {
                        storeGprsOnline(msg, msg.data as GprsOnlineData, ctx)
                    } catch (ex: Exception) {
                        lgr.error("保存三川GPRSOnline失败: ${ex.message}", ex)
                    }
                }

                // 0x00 表示服务器有后续命令， 水表不能下线
                replyPacket(msg, ctx)

                // 提取累计行度
                val range = dataMapper.realtimeDateRange(DataParam(extId = extId).apply {
                    sampleTime2 = Date()
                })

                fetchForward(msg, ctx, recentWeek(range?.maxDateTime))
            }
            CtrlbType.FetchDetailReply -> {
                if (msg.data is FetchDataReplyWeekDetail) {
                    val wd = msg.data as FetchDataReplyWeekDetail

                    GlobalScope.async {
                        try {
                            storeWeekDetail(msg, wd, ctx)
                        } catch (ex: Exception) {
                            lgr.error("保存详细数据异常 ${msg.meterId?.reversedBcdString()}: ${ex.message}", ex)
                        }
                    }

                    // add delta 8 minutes
                    val endTime = wd.startDate!!.plusMinutes(wd.pointNumber!! * wd.timeInterval!! * wd.cmd!!.periodType.minutes)
                    val cntForward = wd.forwardList?.count { it != null } ?: 0
                    val cntPre = wd.pressureList?.count { it != null } ?: 0
                    if ((cntForward == 0 && endTime.plusHours(1).isAfterNow)
                            || endTime.minusHours(9).isAfterNow) {
                        // game over.
                        sendPduList(msg.meterId!!, ctx)
                    } else {
                        // next start-time.
                        if (cntForward < 1) {
                            // fetch forward if last is not forward
                            fetchForward(msg, ctx, endTime)
                        } else {
                            // fetch pressure if last is forward
                            fetchPressure(msg, ctx, wd.startDate!!)
                        }
                    }
                }
            }

            else -> {
                sendPduList(msg.meterId!!, ctx)
            }
        }
    }

    /**
     * send additional pdu list after fetching data.
     */
    private fun sendPduList(eid: BcdString, ctx: ChannelHandlerContext?) {
        val plist = loadPduList(eid.reversedBcdString())
        if (plist.isEmpty()) {
            offlineGameOver(eid, ctx)
            return
        }

        // add game-over to last.
        val reply = SanchuanPacket().apply {
            ctrlb = CtrlbType.Offline
            actrlb = 0
            meterId = eid
            data = MeterOfflineData(null).also {
                it.flag = OfflineFlagType.GameOver
            }
        }
        val buf = ctx!!.alloc().buffer().also { reply.fillBuffer(it) }
        plist.add(ByteArray(buf.readableBytes()).also { buf.readBytes(it) })

        // set to session-attribute.
        AttributeKey.newInstance<ArrayList<ByteArray>>(KEY_PDU_LIST).also {
            ctx.channel()?.attr(it)?.set(plist)
        }
    }

    /**
     * trigger to send additional pdu per time.
     * remove itself if all pdu sent out.
     */
    override fun userEventTriggered(ctx: ChannelHandlerContext?, evt: Any?) {
        if (!AttributeKey.exists(KEY_PDU_LIST)) return
        if (evt !is IdleStateEvent || evt.state() != IdleState.READER_IDLE) return

        val key = AttributeKey.valueOf<ArrayList<ByteArray>>(KEY_PDU_LIST)
        val attr = ctx?.channel()?.attr(key)
        val slist = attr!!.get()
        if (slist?.isNotEmpty() == true) {
            val ba = slist.first()
            slist.removeAt(0)
            attr.set(slist)

            ctx.writeAndFlush(ctx.alloc().buffer().apply { this.writeBytes(ba) })
            lgr.info("Additional PDU [rest: ${slist.size}]: ${BcdString.hexString(ba, true)}")
        }
    }

    internal fun loadPduList(eid: String): ArrayList<ByteArray> {
        val rlist = ArrayList<ByteArray>()

        File(".").listFiles { dir, name ->
            name?.startsWith(eid, true) == true && name.endsWith(".scp", true)
        }.forEach {
            try {
                val ba = BcdString.toByteArray(it.readText(Charsets.UTF_8))
                it.renameTo(File(it.parent, "${it.name}.sent"))

                if (ba != null) rlist.add(ba)
            } catch (ex: Exception) {
                lgr.error("Fail to parse ${it} caused by ${ex.message}", ex)
            }
        }

        return rlist
    }

    /**
     * return the time within this week.
     */
    internal fun recentWeek(maxTime: DateTime?): DateTime {
        val dtNow = DateTime.now()
        return when {
            maxTime == null || maxTime.isAfterNow -> dtNow.minusDays(1).withTimeAtStartOfDay()
            maxTime.isBefore(dtNow.minusDays(1)) -> dtNow.minusDays(1).withTimeAtStartOfDay()
            else -> maxTime.withTimeAtStartOfDay().plusHours(maxTime.hourOfDay / 8 * 8)
        }
    }

    /**
     * store rtu & rtu-log for gprs-online.
     */
    protected fun storeGprsOnline(msg: SanchuanPacket, god: GprsOnlineData, ctx: ChannelHandlerContext?) {
        val extId = "${firmCode}${msg.meterId!!.reversedBcdString()}"
        val txt = BcdString.hexString(msg.backArray!!)

        BwRtuLog().apply {
            this.rtuId = extId
            this.meterId = extId
            this.logTime = Date()
            this.forwardReading = god.totalReading
            this.literPulse = 1000

            this.logLen = msg.len
            this.logText = txt.take(200)
            this.logCmd = "${msg.ctrlb}"
            this.remoteServer = "${ctx?.channel()?.remoteAddress()}"
        }.also {
            it.firmId = firmId
            rtuMapper.addRtuLog(it)
        }

        BwRtu().apply {
            this.rtuId = extId
            this.meterId = extId
            this.forwardReading = god.totalReading
            this.lastDataTime = Date()

            this.lastText = txt.take(200)
            this.lastCmd = "${msg.ctrlb}"
            this.lastTime = Date()

            this.remoteServer = "${ctx?.channel()?.remoteAddress()}"
            this.literPulse = 1000
        }.also {
            it.firmId = firmId
            if (1 > rtuMapper.updateRtu(it)) {
                it.firstTime = Date()
                it.firstCmd = "${msg.ctrlb}"
                it.firstText = txt.take(40)

                rtuMapper.addRtu(it)
            }
        }

        // the reading is invalid maybe!
        var dt = DateTime(god.startDateTime)
        var rcnt = 0
        god.dataList?.forEach {
            // do not persist today.
            if (dt.plusDays(2).isAfterNow) {
                return@forEach
            }

            BwData().apply {
                this.meterId = extId
                this.extId = extId
                this.sampleTime = dt.toDate()
                this.forwardDigits = it
                this.literPulse = 1000
            }.also {
                it.firmId = firmId
                if (1 > dataMapper.updateRealtimeByValue(it))
                    dataMapper.insertRealtime(it)
            }
            ++rcnt

            dt = dt.plusDays(1)
        }

        lgr.info("保存三川${msg.ctrlb} / ${extId} 成功: ${rcnt} / ${JSON.toJSONString(msg, true).take(100)}...")
    }

    /**
     * All realtime data is fetched or fetching 3 times in one session, so offline now.
     */
    fun offlineGameOver(eid: BcdString, ctx: ChannelHandlerContext?) {
        val reply = SanchuanPacket().apply {
            ctrlb = CtrlbType.Offline
            actrlb = 0
            meterId = eid
            data = MeterOfflineData(null).also {
                it.flag = OfflineFlagType.GameOver
            }
        }

        val buf = ctx!!.alloc().buffer().also { reply.fillBuffer(it) }
        lgr.info("提取数据结束 [${eid.reversedBcdString()}]:\n${BcdString.hexString(buf, withBlank = true)}")
        ctx.writeAndFlush(buf)
    }

    /**
     * 保存提取数据
     */
    private fun storeWeekDetail(msg: SanchuanPacket, wd: FetchDataReplyWeekDetail, ctx: ChannelHandlerContext?) {
        val extId = "${firmCode}${msg.meterId!!.reversedBcdString()}"
        val txt = BcdString.hexString(msg.backArray!!, false)

        BwRtuLog().apply {
            rtuId = extId
            meterId = extId

            logTime = Date()
            logText = txt.take(200)
            logLen = logText?.length ?: 0
            remoteServer = ctx?.channel()?.remoteAddress().toString()

            literPulse = 1000
            forwardReading = wd.forwardList?.last()
        }.also {
            it.firmId = firmId
            rtuMapper.addRtuLog(it)
        }

        BwRtu().apply {
            rtuId = extId
            meterId = extId

            lastText = txt.take(200)
            lastTime = Date()
            lastCmd = "WeekDetail"

            literPulse = 1000
            forwardReading = wd.forwardList?.last()
        }.also {
            it.firmId = firmId

            if (1 > rtuMapper.updateRtu(it)) {
                it.firstCmd = "WeekDetail"
                it.firstTime = Date()
                it.firstText = txt.take(40)

                rtuMapper.addRtu(it)
            }
        }

        wd.forwardList.orEmpty().forEachIndexed { idx, fr ->
            if (fr == null) return@forEachIndexed

            BwData().apply {
                meterId = extId
                sampleTime = wd.startDate!!.plusMinutes(idx * wd.timeInterval!! * wd.cmd!!.periodType.minutes).toDate()
                forwardDigits = fr
                literPulse = 1000
            }.also {
                it.extId = extId
                it.firmId = firmId

                if (1 > dataMapper.updateRealtimeByValue(it))
                    dataMapper.insertRealtime(it)
            }
        }

        wd.pressureList.orEmpty().forEachIndexed { idx, pr ->
            if (pr == null) return@forEachIndexed

            BwData().apply {
                meterId = extId
                sampleTime = wd.startDate!!.plusMinutes(idx * wd.timeInterval!! * wd.cmd!!.periodType.minutes).toDate()

                pressure = pr.toDouble()
                pressureDigits = pr.toDouble()
            }.also {
                it.extId = extId
                it.firmId = firmId

                dataMapper.updatePressureByValue(it)
            }
        }

        lgr.info("保存详细数据成功: ${extId}")
    }

    protected fun replyPacket(msg: SanchuanPacket, ctx: ChannelHandlerContext?) {
        when (msg.ctrlb) {
            CtrlbType.Online, CtrlbType.GprsOnline -> {
                // send back online-reply.
                val reply = SanchuanPacket().apply {
                    this.ctrlb = CtrlbType.OnlineReply
                    this.actrlb = 0
                    this.meterId = msg.meterId
                    this.data = MeterOnlineReply(null).also {
                        it.serverTime = Date()
                        it.serverState = ServerStateType.SendCommand
                    }
                }

                ctx!!.alloc().buffer().also {
                    reply.fillBuffer(it)
                    lgr.info("Write back to meter:\n${BcdString.hexString(it, null, true)}")
                    ctx.writeAndFlush(it)
                }
            }

            else -> {
                offlineGameOver(msg.meterId!!, ctx)
            }
        }
    }

    /**
     * 提取累计行度
     */
    private fun fetchForward(msg: SanchuanPacket, ctx: ChannelHandlerContext?, lastDataTime: DateTime) {
        // store fetchTimes
        val key = if (AttributeKey.exists(KEY_FETCH_TIMES))
            AttributeKey.valueOf<Int>(KEY_FETCH_TIMES)
        else
            AttributeKey.newInstance<Int>(KEY_FETCH_TIMES)
        val ft = ctx!!.channel().attr(key)

        // always to avoid fetching.
        lock.writeLock().withLock {
            ft.set(1 + ft.get())

            if (ft.get() > TIMES_PER_SESSION) {
                sendPduList(msg.meterId!!, ctx)

                return
            }
        }

        val fetchPacket = SanchuanPacket().apply {
            this.ctrlb = CtrlbType.FetchDetail
            this.actrlb = 0x02
            this.meterId = msg.meterId
        }

        fetchPacket.data = FetchData(null).apply {
            this.cmdWeekDetail = FetchCmdWeekDetail(0x03)
            this.timeInterval = 1
            this.pointNumber = 48
            this.startTime = lastDataTime
        }

        val buf = ctx.alloc().buffer().apply { fetchPacket.fillBuffer(this) }
        lgr.info("行度提取 [${ft.get()}:${msg.meterId?.reversedBcdString()}]: ${BcdString.hexString(buf, withBlank = true)}")
        ctx.writeAndFlush(buf)
    }

    /**
     * 提取累计行度
     */
    private fun fetchPressure(msg: SanchuanPacket, ctx: ChannelHandlerContext?, lastDataTime: DateTime) {
        // store fetchTimes
        val key = if (AttributeKey.exists(KEY_FETCH_TIMES))
            AttributeKey.valueOf<Int>(KEY_FETCH_TIMES)
        else
            AttributeKey.newInstance<Int>(KEY_FETCH_TIMES)
        val ft = ctx!!.channel().attr(key)

        // always to avoid fetching.
        lock.writeLock().withLock {
            ft.set(1 + ft.get())

            if (ft.get() > TIMES_PER_SESSION) {
                sendPduList(msg.meterId!!, ctx)

                return
            }
        }

        val fetchPacket = SanchuanPacket().apply {
            this.ctrlb = CtrlbType.FetchDetail
            this.actrlb = 0x02
            this.meterId = msg.meterId
        }

        fetchPacket.data = FetchData(null).apply {
            this.cmdWeekDetail = FetchCmdWeekDetail(0x03)
            this.cmdWeekDetail?.dataType = FetchCmdWeekDetail.DetailDataType.Pressure
            this.timeInterval = 1
            this.pointNumber = 48
            this.startTime = lastDataTime
        }

        val buf = ctx.alloc().buffer().apply { fetchPacket.fillBuffer(this) }
        lgr.info("压力提取 [${ft.get()}:${msg.meterId?.reversedBcdString()}]: ${BcdString.hexString(buf, withBlank = true)}")
        ctx.writeAndFlush(buf)
    }

    /**
     * store san-chuan packets.
     */
    protected fun storeOnline(msg: SanchuanPacket, data: MeterOnlineData, ctx: ChannelHandlerContext?) {
        val extId = "${firmCode}${msg.meterId!!.reversedBcdString()}"

        kotlin.run {
            val bd = BwData().apply {
                this.meterId = extId
                this.extId = extId
                this.sampleTime = data.sampleTime

                this.literPulse = 1000
                this.forwardDigits = data.totalReading
                this.revertDigits = data.totalReservseReading

                this.flowMax = data.maxFlow
            }.also { it.firmId = firmId }

            if (1 > dataMapper.updateRealtime(bd)) {
                dataMapper.insertRealtime(bd)
            }
        }

        val text = BcdString.hexString(msg.backArray!!).take(200)

        kotlin.run {
            val rtuLog = BwRtuLog().apply {
                this.rtuId = extId
                this.meterId = extId

                this.literPulse = 1000
                this.forwardReading = data.totalReading
                this.revertReading = data.totalReservseReading

                this.logTime = Date()
                this.logCmd = "${msg.ctrlb}"
                this.logText = text
                this.remoteServer = ctx?.channel()?.remoteAddress()?.toString()

                this.voltWarn = data.volt?.toShort() ?: -1
                this.rssi = data.rsrp!!
                this.cutWarn = data.gmState?.avoidBrokenFlag!!.toShort()
                this.voltWarn = data.gmState?.lowVoltLevelOne!!.toShort()
                this.maxWarn = data.gmState?.highSpeedFlag!!.toShort()
            }.also { it.firmId = firmId }

            rtuMapper.addRtuLog(rtuLog)
        }

        kotlin.run {
            val rtu = BwRtu().apply {
                this.rtuId = extId
                this.meterId = extId

                this.literPulse = 1000
                this.forwardReading = data.totalReading
                this.revertReading = data.totalReservseReading

                this.lastCmd = "SC-Online"
                this.lastText = text
                this.lastTime = Date()
                this.lastDataTime = Date()

                this.voltWarn = data.volt?.toShort() ?: -1
                this.rssi = data.rsrp!!
                this.cutWarn = data.gmState?.avoidBrokenFlag!!.toShort()
                this.voltWarn = data.gmState?.lowVoltLevelOne!!.toShort()
                this.maxWarn = data.gmState?.highSpeedFlag!!.toShort()
            }.also { it.firmId = firmId }

            if (1 > rtuMapper.updateRtu(rtu)) {
                rtuMapper.addRtu(
                        rtu.apply {
                            this.firstCmd = "SC-Online"
                            this.firstText = text.take(40)
                            this.firstTime = Date()
                        }
                )
            }
        }

        lgr.info("Store San-Chuan packets successfully: ${extId} at ${data.sampleTimeStr?.bcdString}")
    }

    /**
     * Calls [ChannelHandlerContext.fireExceptionCaught] to forward
     * to the next [ChannelHandler] in the [ChannelPipeline].
     *
     * Sub-classes may override this method to change behavior.
     */
    override fun exceptionCaught(ctx: ChannelHandlerContext?, cause: Throwable?) {
        lgr.error("错误退出:${ctx?.channel()?.remoteAddress()} - ${cause?.message}", cause)

        ctx?.close()
    }

    companion object {
        private val lgr = LoggerFactory.getLogger(SanchuanTcpHandler::class.java)

        const val KEY_FETCH_TIMES = "fetchTimes"

        const val KEY_PDU_LIST = "pduList"

        const val TIMES_PER_SESSION = 8
    }
}